/*
 * =====================================================================================
 *
 *       Filename:  message_redis_adapter.cc
 *
 *    Description:  defination for redis adapter for IM message.
 *
 *        Version:  1.0
 *        Created:  2013年05月10日 13时42分18秒
 *       Revision:  none
 *       Compiler:  gcc
 *
 *         Author:  Wu Yuxin (wyx, Hin)
 *        Company:  www.Qihoo.net
 *
 * =====================================================================================
 */

#include "message_redis_adapter.h"

#include <stdio.h>
#include <boost/shared_ptr.hpp>
#include <boost/foreach.hpp>
#include <boost/format.hpp>
#include <thrift/Thrift.h>
#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/transport/TBufferTransports.h>

#include "redis_executor/scoped_reply.h"
#include "base/logging.h"
#include "../common/common.h"

using apache::thrift::transport::TTransport;
using apache::thrift::transport::TMemoryBuffer;
using apache::thrift::protocol::TProtocol;
using apache::thrift::protocol::TBinaryProtocol;

typedef boost::shared_ptr<TTransport> Transport_Ptr;
typedef boost::shared_ptr<TProtocol> Protocol_Ptr;

inline int64_t __max(int64_t x, int64_t y) { return (x > y ? x : y); }
inline int64_t __min(int64_t x, int64_t y) { return (x < y ? x : y); }
inline void __safe_free_reply(void *reply) {
  if (NULL != reply)
    freeReplyObject(reply);
}

namespace imserver {

MessageRedisAdapter::MessageRedisAdapter(RedisLocator *locator) : redis_(locator) {
  LOG_INFO("MessageRedisAdapter constructor.");
}

MessageRedisAdapter::~MessageRedisAdapter() {}

int MessageRedisAdapter::AddImMessage(const ImMessage &msg) {
  LOG_INFO("[MessageRedisAdapter::AddImMessage] begin.seq_id-" << msg.seq_id << " sender-" << msg.sender
      << " receiver-" << msg.receiver << " body-" << msg.body << " flags-" << msg.flags);
  // serialize
  Transport_Ptr transport(new TMemoryBuffer(16));
  Protocol_Ptr protocol(new TBinaryProtocol(transport));
  uint32_t size = msg.write(protocol.get());
  uint8_t *serialize = NULL;
  (dynamic_cast<TMemoryBuffer*>(transport.get()))->getBuffer(&serialize, &size);

  // every thing is ok, add the message
  if (NULL == serialize) {
    LOG_ERROR("add message(" << msg.seq_id << ") serialize error.");
    return -2;
  }

  redisContext *context = redis_.AllocContext("");
  if (NULL == context) {
    LOG_ERROR("add message(" << msg.seq_id << ") alloc redis context error.");
    return -2;
  }

  // start a transaction
  // in the transaction, we will do following actions
  // 1. save the message-entity as pair as key/value
  // 2. add the key of message-entity into the message-id-list for sender and receiver
  // 3. check the message is read or not, if not, add it into the unread-message-id-list
  // execute the action
  const std::string read_key = boost::str(boost::format("im:message:read.%1%.%2%") % __min(msg.sender, msg.receiver) % __max(msg.sender, msg.receiver));
  const std::string unread_key = boost::str(boost::format("im:message:unread.%1%") % msg.receiver);
  redisAppendCommand(context, "MULTI");
  redisAppendCommand(context, "SET im:message:%lld %b", msg.seq_id, serialize, size);
  redisAppendCommand(context, "LPUSH %s im:message:%lld", read_key.c_str(), msg.seq_id);
  // check the flags, 0x1 is read 0x0 is unread
  bool unread = false;
  if ((msg.flags & MSG_FLAG_READ) != MSG_FLAG_READ) {
    redisAppendCommand(context, "LPUSH %s im:message:%lld", unread_key.c_str(), msg.seq_id);
    unread = true;
  }
  redisAppendCommand(context, "EXEC");
  int r_count = unread ? 5 : 4;
  redisReply *r = NULL;
  while (r_count-- > 0) {
    if (0 != redisGetReply(context, (void**)&r) || !r) {
      LOG_ERROR("add message(" << msg.seq_id << ") get redis reply error.");
      redis_.ReleaseContext("", context, false);
      return -1;
    }
    freeReplyObject(r);
    r = NULL;
  }
  // release context.
  redis_.ReleaseContext("", context, true);
  return 0;
}

int MessageRedisAdapter::RemoveImMessage(const int64_t sender, const int64_t receiver, const int64_t seq_id) {
  LOG_INFO("[MessageRedisAdapter::RemoveImMessage] begin.seq_id-" << seq_id << " sender-" << sender << " receiver-" << receiver);
  std::string str;
  if (seq_id > 0) {
    str = boost::str(boost::format("LREM im:message:read.%1%.%2% 0 im:message:%3%") % __min(sender, receiver) % __max(sender, receiver) % seq_id);
  } else {
    str = boost::str(boost::format("DEL im:message:read.%1%.%2%") % __min(sender, receiver) % __max(sender, receiver));
  }

  ScopedReply reply(redis_.Execute("", str.c_str()));
  return 0;
}

int MessageRedisAdapter::SetReadImMessage(const int64_t sender, const int64_t receiver) {
  return 0;
}

int MessageRedisAdapter::SetReadImMessageByID(const std::vector<int64_t> & seq_ids, const int64_t uid) {
  LOG_INFO("[MessageRedisAdapter::SetReadImMessageByID] begin. receiver-" << uid);
  std::string mget("MGET");
  redisContext *context = redis_.AllocContext("");
  if (NULL == context) return -2;
  BOOST_FOREACH(const int64_t id, seq_ids) {
    mget.append(boost::str(boost::format(" im:message:%1%") % id));
    redisAppendCommand(context, boost::str(boost::format("LREM im:message:unread.%1% 0 im:message:%2%") % uid % id).c_str());
  }

  redisAppendCommand(context, mget.c_str());
  redisReply *reply = NULL;
  /////////////////////////////////////////////////////////////////////////
  // LREM
  for (size_t i=0; i<seq_ids.size(); ++i) {
    if(0 != redisGetReply(context, (void**)&reply) || !reply) {
      LOG_ERROR("set message read error.(LREM - Get redis reply error)");
      redis_.ReleaseContext("", context, false);
      return -1;
    }
    freeReplyObject(reply);
    reply = NULL;
  }
  /////////////////////////////////////////////////////////////////////////
  // MGET key1 [key2 ...] timeout
  reply = NULL;
  if(0 != redisGetReply(context, (void**)&reply) || !reply) {
    LOG_ERROR("set message read error.(MGET - Get redis reply error.[" << mget << "])");
    redis_.ReleaseContext("", context, false);
    return -1;
  }
  /////////////////////////////////////////////////////////////////////////
  // SET
  // set value back.
  if (reply && reply->type == REDIS_REPLY_ARRAY) {
    for (size_t i=0; i<reply->elements; ++i) {
      ImMessage message;
      if (reply->element[i]->type != REDIS_REPLY_STRING) continue;
      {
        Transport_Ptr transport(new TMemoryBuffer((uint8_t*)reply->element[i]->str, reply->element[i]->len));
        Protocol_Ptr protocol(new TBinaryProtocol(transport));
        message.read(protocol.get());
        message.flags |= MSG_FLAG_READ;
      }
      {
        Transport_Ptr transport(new TMemoryBuffer(16));
        Protocol_Ptr protocol(new TBinaryProtocol(transport));
        uint32_t size = message.write(protocol.get());
        uint8_t *serialize = NULL;
        (dynamic_cast<TMemoryBuffer*>(transport.get()))->getBuffer(&serialize, &size);
        redisAppendCommand(context, "SET im:message:%lld %b", message.seq_id, serialize, size);
      }
    }

    redisReply *reply_set = NULL;
    for (size_t i=0; i<reply->elements; ++i) {
      if (0 != redisGetReply(context, (void**)&reply_set)) {
        LOG_ERROR("set message read error.(SET - Get redis reply failed)");
        redis_.ReleaseContext("", context, false);
        return -1;
      }
      freeReplyObject(reply_set);
      reply_set = NULL;
    }
  }

  freeReplyObject(reply);
  redis_.ReleaseContext("", context, true);
  return 0;
}

int MessageRedisAdapter::GetOfflineImMessages(MessageResult& result, const int64_t receiver, const int32_t offset, const int32_t count) {
  LOG_INFO("[MessageRedisAdapter::GetOfflineImMessages] begin. receiver-" << receiver << " offset-" << offset << "count-" << count);
  std::string key = boost::str(boost::format("im:message:unread.%1%") % receiver);
  return GetMessagesFromList(result, key, offset, count);
}

int MessageRedisAdapter::GetImMessages(MessageResult& result, const int64_t sender, const int64_t receiver, const int32_t offset, const int32_t count) {
  LOG_INFO("[MessageRedisAdapter::GetImMessages] begin. sender-" << sender << " receiver-" << receiver << " offset-" << offset << " count-" << count);
  std::string key = boost::str(boost::format("im:message:read.%1%.%2%") % __min(sender, receiver) % __max(sender, receiver));
  return GetMessagesFromList(result, key, offset, count);
}

int MessageRedisAdapter::GetMessagesFromList(MessageResult& ret, const std::string& key, const int32_t offset, const int32_t count) {
  redisContext *context = redis_.AllocContext("");
  if (NULL == context) {
    LOG_ERROR("get message from list(" << key << ") error.(alloc redis context error)");
    return -2;
  }

  redisAppendCommand(context, "LRANGE %s %d %d", key.c_str(), offset, (offset + count - 1));
  redisAppendCommand(context, "LINDEX %s -1", key.c_str());
  //get result
  redisReply *r_range = NULL;
  redisReply *r_last  = NULL;
  if (REDIS_OK != redisGetReply(context, (void**)&r_range)) {
    LOG_ERROR("get message from list(" << key << ") error.(LRANGE - redis get reply error)");
    redis_.ReleaseContext("", context, false);
    return -1;
  }

  if (REDIS_OK != redisGetReply(context, (void**)&r_last)) {
    LOG_ERROR("get message from list(" << key << ") error.(LINDEX - redis get reply error)");
    redis_.ReleaseContext("", context, false);
    return -1;
  }

  if (!(r_range) || !(r_last)) {
    LOG_ERROR("get message from list(" << key << ") error.(invalid redis reply-[lrange:" << r_range << ", lindex:" << r_last << "])");
    redis_.ReleaseContext("", context, false);
    return -1;
  }

  redis_.ReleaseContext("", context, true);

  int result = 0;
  if (r_range->type == REDIS_REPLY_ARRAY && r_last->type == REDIS_REPLY_STRING) {
    // Get read-message-body.
    // Get serialized strings by keys
    // that from the redis-reply of LRANGE's result.
    // create command list.
    std::string mget("MGET");
    // don't get flags value.
    for (size_t i=0; i<r_range->elements; ++i) {
      if (strcasecmp("message.list.flags", r_range->element[i]->str) == 0) continue;
      mget.append(boost::str(boost::format(" %1%") % r_range->element[i]->str));
    }

    // Deserialized all body-strings.
    if ("MGET" != mget) {
      LOG_DEBUG("redis execute command: " << mget);
      ScopedReply reply(redis_.Execute("", mget.c_str()));
      if (reply && reply->type == REDIS_REPLY_ARRAY) {
        for (size_t i=0; i<reply->elements; ++i) {
          //deserialized and push-back to the list.
          Transport_Ptr transport(new TMemoryBuffer((uint8_t*)reply->element[i]->str, reply->element[i]->len));
          Protocol_Ptr protocol(new TBinaryProtocol(transport));
          ret.results.push_back(ImMessage());
          ret.results.back().read(protocol.get());
        }
      }
    } else {
      LOG_INFO("redis execute command error.(no MGET parameteres)");
    }

    result = (strcasecmp(r_last->str, "message.list.flags") == 0) ? 1 : 0;
    LOG_INFO("get message from list("<<key<<") with flags-" << result);
  } else {
    LOG_INFO("get message from list("<<key<<") invalid reply.");
  }

  freeReplyObject(r_range);
  freeReplyObject(r_last);
  return result;
}

int MessageRedisAdapter::GetOfflineImMessageCount(int32_t * count, const int64_t sender) {
  LOG_INFO("[MessageRedisAdapter::FlagRead] begin.sender-" << sender);
  std::string str = boost::str(boost::format("LLEN im:message:unread.%1%") % sender);
  ScopedReply reply(redis_.Execute("", str.c_str()));
  if (reply && reply->type == REDIS_REPLY_INTEGER) {
    *count = reply->integer;
    return 0;
  } else {
    *count = -1;
  }
  return -1;
}

void MessageRedisAdapter::FlagRead(const int64_t sender, const int64_t receiver) {
  LOG_INFO("[MessageRedisAdapter::FlagRead] begin.sender-" << sender << " receiver-" << receiver);
  std::string key = boost::str(boost::format("im:message:read.%1%.%2%") % __min(sender, receiver) % __max(sender, receiver));
  Flag(key);
}

void MessageRedisAdapter::FlagUnread(const int64_t receiver) {
  LOG_INFO("[MessageRedisAdapter::FlagUnread] begin.receiver-" << receiver);
  std::string key = boost::str(boost::format("im:message:unread.%1%") % receiver);
  Flag(key);
}

inline void MessageRedisAdapter::Flag(const std::string& key) {
  std::string str = boost::str(boost::format("RPUSH %1% message.list.flags") % key);
  ScopedReply reply(redis_.Execute("", str.c_str()));
}

}
